package com.dc.rxjava;

import android.util.Log;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

/**
 * 创建操作符
 *
 * @author 冰柠檬dc
 */
public class CreateOperatorUtils {

    public static final String TAG = "RxJava";

    /**
     * create操作符：创建一个被观察者
     *
     * @return Disposable
     */
    public static Disposable create() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onComplete();
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onComplete
    }

    /**
     * just操作符：创建一个被观察者，并发送事件，发送的事件不可以超过10个以上
     *
     * @return Disposable
     */
    public static Disposable just() {
        return Observable.just(1, 2).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * fromArray操作符：这个方法和just类似，只不过fromArray可以传入多余10个的变量，并且可以传入一个数组
     *
     * @return Disposable
     */
    public static Disposable fromArray() {
        Integer[] array = {1, 2};
        return Observable.fromArray(array).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * fromIterable操作符：直接发送一个Iterable集合数据给观察者
     *
     * @return Disposable
     */
    public static Disposable fromIterable() {
        List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        return Observable.fromIterable(list).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * fromCallable操作符：Callable 是 java.util.concurrent 中的 Callable，Callable 和 Runnable 的用法基本一致，只是它会返回一个结果值，这个结果值就是发给观察者的。
     *
     * @return Disposable
     */
    public static Disposable fromCallable() {
        return Observable.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 1;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onComplete
    }

    /**
     * fromFuture操作符：Future 是 java.util.concurrent 中的 Future，Future 的作用是增加了 cancel() 等方法操作 Callable，它可以通过 get() 方法来获取 Callable 返回的值。
     * doOnSubscribe() 的作用就是只有订阅时才会发送事件
     *
     * @return Disposable
     */
    public static Disposable fromFuture() {
        FutureTask<Integer> futureTask = new FutureTask<>(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 1;
            }
        });
        return Observable.fromFuture(futureTask).doOnSubscribe(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                futureTask.run();
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onComplete
    }

    /**
     * defer操作符：直到被观察者被订阅后才会创建被观察者。
     * defer() 只有观察者订阅的时候才会创建新的被观察者，所以每订阅一次就会打印一次，并且都是打印最新的值。
     *
     * @return Disposable
     */
    public static Disposable defer() {
        return Observable.defer(new Callable<ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> call() throws Exception {
                return Observable.just(1, 2);
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * timer操作符：当到指定时间后就会发送一个 0L 的值给观察者。
     *
     * @return Disposable
     */
    public static Disposable timer() {
        return Observable.timer(2, TimeUnit.SECONDS).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:0
        // onComplete
    }

    /**
     * interval操作符：每隔一段时间就会发送一个事件，这个事件是从0开始，不断增1的数字。
     *
     * @return Disposable
     */
    public static Disposable interval() {
        return Observable.interval(2, 2, TimeUnit.SECONDS).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:0
        // onNext:1
        // onNext:2
        // ......
    }

    /**
     * interval操作符：可以指定发送事件的开始值和数量，其他与 interval() 的功能一样。
     *
     * @return Disposable
     */
    public static Disposable intervalRange() {
        return Observable.intervalRange(2, 3, 2, 1, TimeUnit.SECONDS).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:2
        // onNext:3
        // onNext:4
        // onComplete
    }

    /**
     * range操作符：同时发送一定范围的事件序列。
     * rangLong操作符：作用与 range() 一样，只是数据类型为 Long
     *
     * @return Disposable
     */
    public static Disposable range() {
        return Observable.range(2, 3).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:2
        // onNext:3
        // onNext:4
        // onComplete
    }

    /**
     * empty()：直接发送 onComplete() 事件
     * never()：不发送任何事件
     * error()：发送 onError() 事件
     *
     * @return Disposable
     */
    public static Disposable empty() {
        return Observable.empty().subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // empty打印结果：
        // onSubscribe
        // onComplete

        // never打印结果：
        // onSubscribe

        // error打印结果：
        // onSubscribe
        // onError
    }
}
